/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.tagger;

import maxent.mr.util.CountLines;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.NLineInputFormat;
import org.apache.hadoop.util.Tool;

/**
 *
 * @author hadoop
 */
public class TaggerRunnerNLine extends Configured implements Tool{

    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapred.child.java.opts", "-Xmx2000m");
        JobConf job = new JobConf(conf, MaxentTaggerMR.class);


        Path in = new Path(args[1]);
        Path out = new Path(args[2]);

        

        FileInputFormat.setInputPaths(job, in);
        FileOutputFormat.setOutputPath(job, out);
        job.setJobName("TaggerRunner");
        job.setMapperClass(TaggerMapper.class);
        job.setReducerClass(TaggerReducer.class);
        job.setNumReduceTasks(1);

        job.setInputFormat(NLineInputFormat.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputFormat(TextOutputFormat.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        // setting model path
        job.set("MaxentTagger.model", args[3]);

        // set number lines to split
        int replication = Integer.parseInt(conf.get("dfs.replication"));
        job.setInt("mapred.line.input.format.linespermap", (CountLines.getLines(args[1])/replication)+1);


        JobClient.runJob(job);
        return 0;
    }

}
